#include "ZookeeperUtil.h"
#include "MprpcApplication.h"

#include <semaphore.h>

// 全局的watcher监视器   zkserver给zkclient的发送响应后调用该函数
void global_watcher(_zhandle *zh, int type, int state, const char *path, void *watcherCtx)
{
    if (type == ZOO_SESSION_EVENT) // zkserver发送的消息类型，和会话相关的消息类型
    {
        if (state == ZOO_CONNECTED_STATE) // 表明zkserver和zkclient处于连接成功状态
        {
            sem_t *sem = (sem_t *)zoo_get_context(zh);
            sem_post(sem);
        }
    }
}

ZkClient::ZkClient()
    : m_zhandle(nullptr)
{
}

ZkClient::~ZkClient()
{
    if (m_zhandle != nullptr)
        zookeeper_close(m_zhandle); // 关闭句柄，释放资源
}

void ZkClient::Satrt()
{
    // 加载zkserver的地址
    std::string ip = MprpcApplication::GetInstance().GetConfig().Load("zookeeperIp");
    std::string port = MprpcApplication::GetInstance().GetConfig().Load("zookeeperPort");
    // zookeeper规定格式为：ip:port
    std::string server_add = ip + ":" + port;

    /**
     * zookeeper_mt：多线程版本，提供3个线程
     * API调用线程，即目前所在线程
     * 网络IO线程  底层用的是poll。zookeeper_init会创建网络IO线程，专门发起IO连接
     * watcher回调线程。 当zkclient收到zkserver的响应后 创建watcher回调线程并执行回调
     */
    /**
     * 这里调用zookeeper_init创建句柄，建立会话。会话的建立是异步的，zookeeper返回并不意味着会话建立成功，
     * 直到zkserver给zkclient发送响应调用global_watcher函数且满足两个if，会话才建立成功
     */
    m_zhandle = zookeeper_init(server_add.c_str(), global_watcher, 30000, nullptr, nullptr, 0);
    if (m_zhandle == nullptr)
    {
        // std::cout << "zookeeper_init error" << std::endl;
        LOG_ERROR_("zookeeper_init error");
        exit(EXIT_FAILURE);
    }
    // zookeeper_init返回后，并不意味着zkserver给我们发送了响应，只是表明m_zhandle句柄在本地内存开辟初始化成功了

    sem_t sem;
    sem_init(&sem, 0, 0);
    // 给句柄绑定一个信号量
    zoo_set_context(m_zhandle, &sem);

    // 信号量初始为0，肯定在此阻塞，直到zkserver给zkclient发送响应调用global_watcher函数，给信号量+1，才会继续往下执行
    sem_wait(&sem);
    // std::cout << "zookeeper_init success" << std::endl;
    LOG_INFO_("zookeeper_init success");
}

void ZkClient::Create(const char *path, const char *data, int datalen, int state)
{
    char path_buffer[128];
    int bufferlen = sizeof(path_buffer);
    int flag;
    // 同步检查 要创建的节点path在zookeeper中是否存在
    flag = zoo_exists(m_zhandle, path, 0, nullptr);
    if (flag == ZNONODE) // 节点不存在
    {
        flag = zoo_create(m_zhandle, path, data, datalen, &ZOO_OPEN_ACL_UNSAFE, state, path_buffer, bufferlen);
        if (flag == ZOK)
        {
            // std::cout << "znode create success, path:" << path << std::endl;
            LOG_INFO_("znode create success, path:%s", path);
        }
        else
        {
            // std::cout << "zoo_create return flag:" << flag << std::endl;
            // std::cout << "znode create error, path:" << path << std::endl;
            LOG_ERROR_("zoo_create error return flag:%d path:%s", flag, path);
            exit(EXIT_FAILURE);
        }
    }
}

// 根据指定的path，获取节点的数据
std::string ZkClient::GetData(const char *path)
{
    char buffer[64];
    int bufferlen = sizeof(buffer);
    int flag = zoo_get(m_zhandle, path, 0, buffer, &bufferlen, nullptr);
    if (flag == ZOK)
    {
        return buffer;
    }
    else
    {
        // std::cout << "zoo_get return flag:" << flag << std::endl;
        // std::cout << "zoo_get error, path:" << path << std::endl;
        LOG_ERROR_("zoo_get error return flag:%d path:%s", flag, path);
        return "";
    }
}